package com.hudson.broker;

import com.hudson.broker.client.NameSrvClient;
import com.hudson.broker.handler.HandlerProcesserManage;
import com.hudson.broker.config.BrokerConfig;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.handler.netty.BrokerHandler;
import com.hudson.broker.handler.netty.IdleMessageHandler;
import com.hudson.broker.handler.netty.ProcesserHandler;
import com.hudson.codec.MessageCodecFactory;
import com.hudson.core.base.BaseComponent;
import com.hudson.store.dispatch.CommitLogDispatchService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class BrokerController extends BaseComponent {
    private static final Logger logger = LoggerFactory.getLogger(BrokerController.class);

    public static final CountDownLatch startCountDownLatch = new CountDownLatch(1);


    private static final ServerBootstrap serverBootstrap = new ServerBootstrap().group(ThreadPoolHelper.BROKER_BOSS_EVENT_LOOP, ThreadPoolHelper.BROKER_WORK_EVENT_LOOP)
            .channel(NioServerSocketChannel.class)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel sc) throws Exception {
                    sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
                    sc.pipeline().addLast(MessageCodecFactory.getCodec(BrokerConfig.serializableMethod));
                    sc.pipeline().addLast(new IdleStateHandler(30, 0, 0, TimeUnit.MINUTES));
                    sc.pipeline().addLast(new IdleMessageHandler());
                    // 第一个handler，接收到请求时返回信息，告诉发送者发送成功
                    sc.pipeline().addLast(new BrokerHandler());
                    sc.pipeline().addLast(new ProcesserHandler());
//                    sc.pipeline().addLast(new QueueMapHandler());
//                    sc.pipeline().addLast(new DefaultProduceHandler());
//                    sc.pipeline().addLast(new RelayProduceHandler());
//                    sc.pipeline().addLast(new ConsumerToBrokerHandler());
//                    sc.pipeline().addLast(new AddTopicQueueHandler());
                }
            });

    private static Channel brokerChannel;

    public static void start() {
        startStopWatch();

        createBrokerServe();

        HandlerProcesserManage.start();

        NameSrvClient.start();

        startSendBrokerInfo();

        CommitLogDispatchService.start();

        startCountDownLatch.countDown();

        logger.info("Broker 启动成功,耗時:{} ms",getTimeByS());
    }

    public static void createBrokerServe() {
        try {

            brokerChannel = serverBootstrap
                    .bind(new InetSocketAddress(BrokerConfig.BROKER_BIND_ADDRESS, BrokerConfig.BROKER_PORT))
                    .sync()
                    .channel();

            logger.info("Broker已启动，监听地址: {}", BrokerConfig.BROKER_BIND_ADDRESS + ":" + BrokerConfig.BROKER_PORT);

        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void startSendBrokerInfo() {
        ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.scheduleAtFixedRateByWatch(() -> {
            try {
                NameSrvClient.sendBrokerAddress();
            } catch (Exception e) {
                logger.error(e.getMessage());
            }
        }, 0, 30, TimeUnit.SECONDS);
    }

    public static void shutDown() {

        /**
         * 关闭NameSrv的客户端连接
         */
        NameSrvClient.close();

        /**
         * 关闭Broker服务
         */
        try {
            brokerChannel.close().sync().addListener(f -> {
                if (f.isSuccess()) {
                    logger.info("broker 关闭成功");
                } else {
                    logger.error("broker 关闭异常");
                }
            });
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }


        /**
         * 关闭定时线程池和普通自定义线程池，并关闭netty的eventLoop
         */
        ThreadPoolHelper.shutDown();
    }
}
